package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseApp;
import com.atguigu.gmall.realtime.bean.UserLoginBean;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.function.DorisMapFunction;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.DorisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/8/9 10:16
 */
public class Dws_04_DwsUserUserLoginWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_04_DwsUserUserLoginWindow().start(
            40004,
            2,
            "Dws_04_DwsUserUserLoginWindow",
            GmallConstant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
        // 1. 过滤出登录记录
        SingleOutputStreamOperator<JSONObject> loginLogStream = filterLoginLog(stream);
        
        // 2. 解析为 pojo
        SingleOutputStreamOperator<UserLoginBean> beanStream = parseToPojo(loginLogStream);
        
        // 3. 开窗聚合
        SingleOutputStreamOperator<UserLoginBean> resultStream = windowAndAgg(beanStream);
        
        // 4. 写出到 doris
        writeToDoris(resultStream);
    }
    
    private void writeToDoris(SingleOutputStreamOperator<UserLoginBean> resultStream) {
        resultStream
            .map(new DorisMapFunction<>())
            .sinkTo(DorisUtil.getDorisSink("gmall2023.dws_user_user_login_window"));
    }
    
    private SingleOutputStreamOperator<UserLoginBean> windowAndAgg(
        SingleOutputStreamOperator<UserLoginBean> beanStream) {
        return beanStream
            .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<UserLoginBean>() {
                    @Override
                    public UserLoginBean reduce(UserLoginBean value1, UserLoginBean value2) throws Exception {
                        value1.setUuCt(value1.getUuCt() + value2.getUuCt());
                        value1.setBackCt(value1.getBackCt() + value2.getBackCt());
                        return value1;
                    }
                },
                new ProcessAllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>() {
                    @Override
                    public void process(Context ctx,
                                        Iterable<UserLoginBean> values,
                                        Collector<UserLoginBean> out) throws Exception {
                        UserLoginBean bean = values.iterator().next();
                        
                        bean.setStt(DateFormatUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(ctx.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.tsToDate(ctx.window().getStart()));
                        
                        out.collect(bean);
                        
                    }
                }
            );
        
    }
    
    private SingleOutputStreamOperator<UserLoginBean> parseToPojo(
        SingleOutputStreamOperator<JSONObject> loginLogStream) {
        return loginLogStream
            .keyBy(obj -> obj.getJSONObject("common").getString("uid"))
            .process(new KeyedProcessFunction<String, JSONObject, UserLoginBean>() {
                
                private ValueState<String> lastLoginDateState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    // ttl
                    lastLoginDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastLoginDate", String.class));
                }
                
                @Override
                public void processElement(JSONObject value,
                                           Context ctx,
                                           Collector<UserLoginBean> out) throws Exception {
                    
                    String lastVisitDate = lastLoginDateState.value();
                    Long ts = value.getLong("ts");
                    String today = DateFormatUtil.tsToDate(ts);
                    
                    long uuCt = 0;
                    long backCt = 0;
                    if (!today.equals(lastVisitDate)) { // 改用户今天第一登录
                        uuCt = 1;
                        lastLoginDateState.update(today);
                        // 状态值不会null, 则证明以前登录过, 判断是否为回流用户
                        if (lastVisitDate != null) {
                            if (ts - DateFormatUtil.dateToTs(lastVisitDate) > GmallConstant.SEVEN_DAY_MS) {
                                backCt = 1;
                            }
                        }
                        
                    }
                    
                    if (uuCt == 1) {
                        out.collect(new UserLoginBean(
                            "", "",
                            "",
                            backCt, uuCt,
                            ts
                        ));
                    }
                }
            });
    }
    
    private SingleOutputStreamOperator<JSONObject> filterLoginLog(DataStreamSource<String> stream) {
        return stream
            .map(JSON::parseObject)
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((obj, ts) -> obj.getLong("ts"))
                    .withIdleness(Duration.ofSeconds(60))
            ) // 后面数据减少之后, 水印更新就会比较慢
            .filter(new FilterFunction<JSONObject>() {
                @Override
                public boolean filter(JSONObject value) throws Exception {
                    String uid = value.getJSONObject("common").getString("uid");
                    String lastPageId = value.getJSONObject("page").getString("last_page_id");
                    
                    return ("login".equals(lastPageId) || null == lastPageId) && null != uid;
                }
            });
    }
}
/*
统计七日回流用户和当日独立用户数

1. 数据源
    页面日志
    
2. 找到已经登陆的用户
    找到同一个用户当天的第一条登录记录
    
    首先: 过滤登录记录
        两种登录:
            自动登录
                last_page_id = null and uid != null
            
            手动登录
                A -> 登录页 -> B(找到 b)
                
                last_page_id = login and uid != null
                
            合并条件:
                (last_page_id = null || = login ) && uid != null
            
    
    再: 在找用户的当天第一条
        
        使用状态
        
        再: 判断是否为回流用户
            当时第一条登录
                用今天的毫秒值 - 状态中的毫秒值 > 7 天的毫秒值
  
 开窗聚和
    windowAll
 写出到 doris

 */